package util.ndl.v1

import java.util

import grpc.v1.rpcV1.ColValue.Of._
import grpc.v1.rpcV1._
import org.apache.calcite._
import org.apache.calcite.linq4j._
import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory}
import org.apache.calcite.rex._
import org.apache.calcite.schema._
import org.apache.calcite.schema.impl._


class GrpcSchemaFactory extends SchemaFactory {


  override def create(parentSchema: SchemaPlus, name: Str, operand: util.Map[Str, AnyRef]): Schema =

    new AbstractSchema {
      val channel = io.grpc.ManagedChannelBuilder
        .forAddress("localhost", 50051)
        .usePlaintext
        .build()

      def stub = SchemaDataGrpc.blockingStub(channel)

      import scala.collection.convert.WrapAsJava._

      override protected def getTableMap = {
        mapAsJavaMap(stub.getTableInfos(GetTableInfosReq(name)).map { tbl =>
          tbl.name -> new AbstractTable with FilterableTable {
            override def scan(root: DataContext, filters: util.List[RexNode]): Enumerable[Array[AnyRef]] =
              new AbstractEnumerable2[Array[AnyRef]] {

                import ScanFilterVal.{Kind => FilterKind}

                def convert2oper(node: RexNode): ScanFilterVal = node match {
                  case call: RexCall => ScanFilterVal(FilterKind.Filter(convert2filter(call)))
                  case inputRef: RexInputRef =>
                    //colIndex (inputRef.getName.drop(1).toInt) //FIXME: toColName
                    ScanFilterVal(FilterKind.ColName(tbl.cols(inputRef.getName.drop(1).toInt).name))
                  case literal: RexLiteral =>
                    def convert = literal.getValue2.asInstanceOf[Any] match {
                      case str: S => StringValue(str)
                      case i: I => Int32Value(i)
                      case d: D => DoubleValue(d)
                      case l: L => Int64Value(l)
                      case b: B => BoolValue(b)
                      case _ => ???
                    }

                    ScanFilterVal(FilterKind.OperVal(ColValue(convert)))
                }

                def convert2filter(filter: RexNode): ScanFilter = filter match {
                  case filter: RexCall =>
                    ScanFilter(SqlKind.fromName(filter.getKind.name()).getOrElse(???), filter.operands.map {
                      convert2oper
                    })
                }

                override def iterator(): util.Iterator[Array[AnyRef]] =
                  stub.scan(ScanReq(name, tbl.name, filters.map {
                    convert2filter
                  })).map { row =>
                    val data = row.vals.map { t => t.of.value.asInstanceOf[AnyRef] }.toArray
                    data
                  }
              }

            override def getRowType(typeFactory: RelDataTypeFactory): RelDataType = {
              val fieldBuilder = typeFactory.builder()
              tbl.cols.foreach { col =>
                import org.apache.calcite.sql.`type`.SqlTypeName
                fieldBuilder.add(col.name, SqlTypeName.get(col.`type`.name))
              }

              typeFactory.createStructType(fieldBuilder)
            }

          }
        }.toMap)

      }


    }


}
